package DataStreamApi.Transformation算子.物理分区算子;


import domain.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;
import java.util.List;

/**
 * 打乱重分区
 *
 * 在Flink中，一旦数据流被转换为KeyedStream后，就不能再应用如shuffle()这样的物理分区操作了。
 * 这是因为KeyedStream已经根据键（key）对数据进行了逻辑分区，而这些物理分区的操作（如shuffle、rebalance等）
 * 是用于在未分组的数据流上进行更细粒度的控制。
 */
public class Flink01_SHUFFLE {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        List<WaterSensor> waterSensorList = Arrays.asList(
                new WaterSensor("s1", 1L, 1),
                new WaterSensor("s2", 2L, 2),
                new WaterSensor("s2", 3L, 3),
                new WaterSensor("s3", 1L, 4),
                new WaterSensor("s1", 11L, 5),
                new WaterSensor("s1", 2L, 6));
        executionEnvironment.setParallelism(10);

        DataStreamSource<WaterSensor> waterSensorDataStreamSource = executionEnvironment.fromCollection(waterSensorList);


        /**
         * 按照ID分组
         * KeyBY不是转换算子，只是对数据进行了重分区，不能设置并行度。
         *
         * 1、KeyBy是对数据进行分组，保证相同key的数据在同一个分区
         * 2、分区是指将数据划分到不同的区域，一个子任务可以认为是一个分区。
         *
         */
        KeyedStream<WaterSensor, String> waterSensorStringKeyedStream = waterSensorDataStreamSource.keyBy(new KeySelector<WaterSensor, String>() {
            @Override
            public String getKey(WaterSensor waterSensor) throws Exception {
                return waterSensor.getId();
            }
        });




        // 假设waterSensorStringKeyedStream是已经存在的KeyedStream
        DataStream<WaterSensor> dataStreamAfterKeyBy = waterSensorStringKeyedStream
                .map(new MapFunction<WaterSensor, WaterSensor>() {
                    @Override
                    public WaterSensor map(WaterSensor value) throws Exception {
                        // 这里可以不做任何改动，只是简单返回输入值
                        return value;
                    }
                });

// 现在，dataStreamAfterKeyBy是一个普通的DataStream，可以对其应用shuffle等物理分区操作
        DataStream<WaterSensor> shuffleDataStream = dataStreamAfterKeyBy.shuffle();
        shuffleDataStream.print();



        executionEnvironment.execute();

    }
}
